package cn.hesion;
/**
 * ClassName: JavaStreamHandle <br/>
 * Description: <br/>
 * date: 2021/9/7 20:13<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Java Flink 流处理处理入门代码
 * @author: hesion
 * @create: 2021-09-07 20:13
 **/
public class JavaStreamHandle {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> data = env.socketTextStream("tmy", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMaped = data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] word = s.split(",");
                for (String s1 : word) {
                    Tuple2<String, Integer> stringIntegerTuple2 = new Tuple2<>(s1, 1);
                    collector.collect(stringIntegerTuple2);
                }
            }
        });
        KeyedStream<Tuple2<String, Integer>, Tuple> sum = flatMaped.keyBy(0);
        sum.print();

        env.execute();


    }
}
